package com.arch.flink.state;

import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author pizhihui
 * @date 2024-05-09 11:50
 */
public class StateDemo {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        CheckpointConfig ckpConfig = env.getCheckpointConfig();

        env.setStateBackend(new HashMapStateBackend());
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // set mode to exactly-once (this is the default)
        ckpConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // make sure 500 ms of progress happen between checkpoints
        ckpConfig.setMinPauseBetweenCheckpoints(500);

        // checkpoints have to complete within one minute, or are discarded
        ckpConfig.setCheckpointTimeout(60000);

        // only two consecutive checkpoint failures are tolerated
        ckpConfig.setTolerableCheckpointFailureNumber(2);

        // allow only one checkpoint to be in progress at the same time
        ckpConfig.setMaxConcurrentCheckpoints(1);

        // enable externalized checkpoints which are retained
        // after job cancellation
        ckpConfig.setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // enables the unaligned checkpoints
        ckpConfig.enableUnalignedCheckpoints();


        ckpConfig.setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://chek..."));


//        // sets the checkpoint storage where checkpoint snapshots will be written
//        Configuration config = new Configuration();
//        config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
//        config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "hdfs:///my/checkpoint/dir");
//        env.configure(config);
//
//        // enable checkpointing with finished tasks
//        Configuration config = new Configuration();
//        config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
//        env.configure(config);


    }

}
